// @Author EthanScriptOn
// @Desc
package extend_parser

import (
	"fmt"
	"gitee.com/fatzeng/srf_switch_expansion_package/extend_config"
	"gitee.com/fatzeng/srf_switch_stream_bridge/bridge"
	"gitee.com/fatzeng/srf_switch_stream_bridge/framework/actuator"
	"gitee.com/fatzeng/srf_switch_stream_bridge/result_parser"
	"github.com/confluentinc/confluent-kafka-go/kafka"
	"reflect"
)

type KafkaConsumeResponseParser struct {
	result_parser.BytesParser
}

func (k *KafkaConsumeResponseParser) Parse(ctx *actuator.SrfSwitchBridgeActuatorContext, config *bridge.BridgeConfig, response interface{}) (interface{}, error) {
	if !config.ResultType.AssignableTo(reflect.TypeOf(&extend_config.KafkaConsumeResponse{})) {
		return nil, fmt.Errorf("result type is not *extend_config.KafkaConsumeResponse")
	}
	resp := new(extend_config.KafkaConsumeResponse)
	switch e := response.(type) {
	case *kafka.Message:
		resp.IsOk = true
		resp.Msg = string(e.Value)
		resp.Offset = int64(e.TopicPartition.Offset)
		resp.Partition = e.TopicPartition.Partition
		return resp, nil
	case kafka.Error:
		resp.IsOk = false
		resp.ErrMsg = fmt.Sprintf("Received message with exception: %v", e)
		return resp, nil
	default:
		resp.IsOk = false
		resp.ErrMsg = fmt.Sprintf("Unknown message type:%v", e)
		return resp, nil
	}
}
